package org.ala.tiktools.rpc.netty.handler;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rpc.config.RpcConfigInfo;
import org.ala.tiktools.rpc.msg.RpcMsg;
import org.ala.tiktools.rpc.msg.ping.RpcPingMsg;
import org.ala.tiktools.rpc.msg.terminate.RpcTerminateMsg;
import org.slf4j.Logger;

import java.util.concurrent.TimeUnit;

/**
 * 空闲检测
 * <p>  收到任何消息都判定为远端还活着
 * <p>  超过idle间隔没有收到消息，发ping
 * <p>  连续times次间隔没有收到消息，判定远端死亡。断掉链接
 *
 * @author ala
 * @date 2025-04-09 11:37
 */
public class RpcIdleInterceptor extends ChannelInboundHandlerAdapter {


    static Logger log = LoggerFactory.getLogger("rpc", "handler");


    /**
     *  上一次收到消息时间戳
     *  <p> 任何消息都算
     */
    protected long lastTimestamp = -1;
    /**
     *  连续未收到ack间隔次数
     */
    protected int times = 0;
    /**
     *  是否已经追加定时任务
     */
    protected boolean schedule = false;
    /**
     *  是否主动发起心跳
     *  <p> 当超过心跳间隔没有收到ack时，主动发送一次心跳
     */
    protected boolean ping = false;


    public RpcIdleInterceptor(boolean ping) {
        this.ping = ping;
    }


    @SuppressWarnings("rawtypes")
	@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.debug("[RpcIdleInterceptor channelRead] 收到消息:{}", ((RpcMsg)msg).getHead());

        //  收到任何消息，判定远端还活着
        lastTimestamp = System.currentTimeMillis();
        times = 0;
        super.channelRead(ctx, msg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        lastTimestamp = System.currentTimeMillis();
        times = 0;
        if (!schedule) {
            schedule = true;
            ctx.channel().eventLoop().schedule(new IdleTask(ctx, this), RpcConfigInfo.INSTANCE.getIdleInterval(), TimeUnit.MILLISECONDS);
        }
        super.channelActive(ctx);
    }
    public static class IdleTask implements Runnable {
        protected ChannelHandlerContext ctx;
        protected RpcIdleInterceptor idle;

        public IdleTask(ChannelHandlerContext ctx, RpcIdleInterceptor idle) {
            this.ctx = ctx;
            this.idle = idle;
        }

        @Override
        public void run() {
            long now = System.currentTimeMillis();
            boolean again = true;
            //  如果超过时间还未收到消息，主动发一次ping
            if (now - idle.lastTimestamp > RpcConfigInfo.INSTANCE.getIdleInterval()) {
                if (idle.times >= RpcConfigInfo.INSTANCE.getIdleTimesClose()) {
                    try {
                        again = false;
                        ctx.channel().writeAndFlush(RpcTerminateMsg.build());
                        ctx.channel().close();
                        log.info("[RpcIdleInterceptor IdleTask] 超过间隔次数未收到消息 判定为远端已无效，关闭连接 idleInterval:{} idleTimesClose:{} remote:{}", RpcConfigInfo.INSTANCE.getIdleInterval(), RpcConfigInfo.INSTANCE.getIdleTimesClose(), ctx.channel().remoteAddress());
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }
                idle.times++;
                if (idle.ping) {
                    RpcPingMsg m = RpcPingMsg.build();
                    m.getHead().id(-idle.times);
                    ctx.channel().writeAndFlush(m);
                }
            }

            if (again) {
                ctx.channel().eventLoop().schedule(this, RpcConfigInfo.INSTANCE.getIdleInterval(), TimeUnit.MILLISECONDS);
            }
        }
    }

}
